package com.decorator.rocketmq.controller;


import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("rocketmq")
public class RocketmqController {

    Logger log = LoggerFactory.getLogger(RocketmqController.class);

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @RequestMapping("/pushMessage.action")
    public void get(@RequestParam("id") int id) throws Exception {

        DefaultMQProducer producer = new
                DefaultMQProducer("please_rename_unique_group_name");
        // Specify name server addresses.
        producer.setNamesrvAddr("127.0.0.1:9876");
        //Launch the instance.
        producer.start();
        for (int i = 0; i < 100; i++) {
            //Create a message instance, specifying topic, tag and message body.
            Message msg = new Message("orderPay" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ " +
                            i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
            //Call send message to deliver message to one of brokers.
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }
        //Shut down once the producer instance is not longer in use.
        producer.shutdown();


    }


    @RequestMapping("/asyncSend")
    public void asyncSend() throws Exception {

        for (int i = 0; i < 100; i++) {

            int finalI = i;
            rocketMQTemplate.asyncSend("asyncSendTopic", "hollo asyncSendTopic" + i, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println("异步发送成功：" + finalI);
                }

                @Override
                public void onException(Throwable throwable) {
                    System.out.println("异步发送想失败：" + finalI);
                    log.error("async 失败",throwable);
                    System.out.println(throwable.getCause().getMessage());
                }
            });
        }
    }

}
